package org.example.using;


import org.apache.log4j.Level;
import org.apache.log4j.Logger;
import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.FilterFunction;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.SparkSession;
import org.example.bean.*;
import org.example.utils.MysqlUtils_2;

import java.io.IOException;

import static org.apache.spark.sql.functions.col;


/***
 * 创建一个类,用来测试 spark 2.3.0版本的 DataFrame的用法, DataSet API的使用.
 */
public class Spark02_SparkDataSetFilterUsing {
    public static void main(String[] args) throws IOException {
        // 初始化SparkSession
        // 引入log4j,消除控制台的警告
        Logger.getLogger("org").setLevel(Level.WARN);
        Logger.getLogger("akka").setLevel(Level.WARN);
        SparkConf conf = new SparkConf().setAppName("PantographModelFunction") // 设置应用程序名称
                .setMaster("local[*]") // 设置运行模式，这里是本地模式
                .set("spark.executor.memory", "6g") // 设置executor内存大小
                .set("spark.serializer", "org.apache.spark.serializer.KryoSerializer") // 设置序列化器
                .set("spark.driver.allowMultipleContexts", "true") // 允许多个SparkContext同时存在
                .set("spark.hadoop.validateOutputSpecs", "false") // 禁止输出路径已经存在时抛出异常
                .set("hive.mapred.supports.subdirectories", "true") // 开启Hive支持
                .set("mapreduce.input.fileinputformat.input.dir.recursive", "true") // 开启递归读取文件夹中的文件
                .set("spark.debug.maxToStringFields", "100").set("spark.sql.codegen.wholeStage", "false")
                .set("spark.sql.caseSensitive", "false"); // Spark会自动处理大小写不匹配的情况
        JavaSparkContext sc = new JavaSparkContext(conf);
        SparkSession spark = SparkSession.builder().config(conf).getOrCreate();

        /*
            TODO 读取MySQL中的表,根据表结构创建DF.这里需要提前手动创建一个 Java Bean,这里使用MySQL中的一张表进行演示.表中的字段全部转化为小写.
                 演示 `filter()` 方法的用法,这里filter可以有以下几种写法
                 1、使用filter()方法和lambda表达式
                 2、使用filter()方法和Column表达式：
                 3、 使用where()方法和 Column表达式（where()是filter()的别名，因此它们的行为相同)
                 4、针对过滤条件有多个,并且逻辑上涉及到 and 、or 等的逻辑处理上的写法

         */
        Dataset<AdsPhmFault> adsPhmFaultDF = MysqlUtils_2.readMySQLTableDF(spark, "ads_phm_fault", AdsPhmFault.class);
        adsPhmFaultDF.show(100);
        // 第一种写法, filter()方法和`lambda`表达式  通过 `S_TRAIN_ID`字段过滤 `HXD1D-1030`的数据.
        Dataset<AdsPhmFault> filterWithSTrainIdDS = adsPhmFaultDF.filter((FilterFunction<AdsPhmFault>) adsPhmFault -> adsPhmFault.getS_TRAIN_ID().equals("HXD1D-1030"));
        // 如果有多个过滤条件,则可以使用下面的逻辑运算符来进行连接
        Dataset<AdsPhmFault> adsPhmFaultDataset = adsPhmFaultDF.filter(
                (FilterFunction<AdsPhmFault>) adsPhmFault -> adsPhmFault.getPhm_lb().equals("变流器元件故障报警")
                        && adsPhmFault.getPhm_bjwz().equals("1")
                        || adsPhmFault.getPhm_bjwz().equals("2")
        );
        filterWithSTrainIdDS.show(false);
        // 上面方式的全写,不使用 lambda 表达式.不省略写法
        adsPhmFaultDF.filter(new FilterFunction<AdsPhmFault>() {
            @Override
            public boolean call(AdsPhmFault adsPhmFault) throws Exception {
                return adsPhmFault.getS_TRAIN_ID().equals("HXD1D-1030");
            }
        }).show();
        // 第二种写法:直接使用 col. 如果有多个过滤条件,可以按照下面的写法来写,通过and、or进行连接.
        Dataset<AdsPhmFault> filteredDF = adsPhmFaultDF.filter(col("S_TRAIN_ID").equalTo("HXD1D-1001"));
        filteredDF.show();
        adsPhmFaultDF.filter(
                col("S_TRAIN_ID").equalTo("HXD1D-1001")
        );

        // 第三种写法:使用where()方法和 Column表达式（where()是filter()的别名，因此它们的行为相同)
        adsPhmFaultDataset.where(col("phm_lb").equalTo("变流器元件故障报警")
                .and(col("phm_gzmc").equalTo("L1元件总故障, L1A相下管故障"))
                .or(col("phm_gzmc").equalTo("ACC元件总故障"))
        ).show(false);
        spark.stop();
    }
}
